package cn.com.bluemoon.bd.flink.sink;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.HConstants;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.TableName;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.client.Connection;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.client.Delete;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.client.Table;
import org.apache.flink.hbase.shaded.org.apache.hadoop.hbase.util.Bytes;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;


import java.util.Map;
import java.util.function.Function;

public class HBaseRowDeleteSink<IN> extends RichSinkFunction<IN> {
    private Connection conn;
    private String tableName; //hbase的表名
    private Table table;
    private Function<IN, String> rowKeyGenerator;  //rowKey的生成方法
    private static String FAMILY = "info";
    private org.apache.hadoop.conf.Configuration conf;

    public HBaseRowDeleteSink(String tableName, Function<IN, String> rowKeyGenerator) {
        this.tableName = tableName;
        this.rowKeyGenerator = rowKeyGenerator;
    }

    /**
     * 初始化方法
     *
     * @param parameters
     * @throws Exception
     */
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String zkServers = ParameterTool.fromPropertiesFile(HBaseRowDeleteSink.class.getClassLoader().getResourceAsStream("zk.properties")).get("zk.servers");
        conf = new org.apache.hadoop.conf.Configuration();
        conf.set("hbase.zookeeper.quorum", zkServers);
        conf.set("hbase.client.retries.number", "5");
        conf.set("hbase.client.operation.timeout", "30000");
        conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 30000);
        conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 30000);
        buildConnection();
    }

    public void buildConnection() throws Exception {
        conn = ConnectionFactory.createConnection(conf);
        TableName tableName = TableName.valueOf(this.tableName);
        table = conn.getTable(tableName);
    }


    public void invoke(IN value, Context context) throws Exception {
        Delete delete = new Delete(Bytes.toBytes(rowKeyGenerator.apply(value)));

        try {
            table.delete(delete);
        } catch (Exception e) {
            closeConnAndTable();

            try {
                buildConnection();
                table.delete(delete);
            } catch (Exception e1) {
                closeConnAndTable();
                buildConnection();
                throw new Exception("连接hbase错误", e1);
            }
        }
    }

    public void close() throws Exception {
        super.close();
        closeConnAndTable();
    }

    public void closeConnAndTable() {
        try {
            if (table != null) {
                table.close();
            }
        } catch (Exception e) {

        }

        try {
            if (conn != null) {
                conn.close();
            }
        } catch (Exception e) {

        }
    }
}
